package server

import (
	"fmt"
	"net"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"time"

	"gitee.com/zfd81/dbrs/log"

	"gitee.com/zfd81/dbrs/errs"

	"gitee.com/zfd81/dbrs/engine"

	"gitee.com/zfd81/dbrs/sql/parser"

	"gitee.com/zfd81/dbrs/sys"
	"vitess.io/vitess/go/mysql/collations"

	"gitee.com/zfd81/dbrs/config"
	"gitee.com/zfd81/dbrs/expr"
	"github.com/shopspring/decimal"
	"github.com/spf13/cast"

	"github.com/opentracing/opentracing-go"

	"github.com/go-kit/kit/metrics/discard"
	"github.com/sirupsen/logrus"
	"gopkg.in/src-d/go-errors.v1"
	"vitess.io/vitess/go/mysql"
	"vitess.io/vitess/go/netutil"
	"vitess.io/vitess/go/sqltypes"
	querypb "vitess.io/vitess/go/vt/proto/query"
)

var regKillCmd = regexp.MustCompile(`^kill (?:(query|connection) )?(\d+)$`)

var errConnectionNotFound = errors.NewKind("connection not found: %c")

// ErrRowTimeout will be returned if the wait for the row is longer than the connection timeout
var ErrRowTimeout = errors.NewKind("row read wait bigger than connection timeout")

// ErrConnectionWasClosed will be returned if we try to use a previously closed connection
var ErrConnectionWasClosed = errors.NewKind("connection was closed")

var ErrUnsupportedOperation = errors.NewKind("unsupported operation")

type Handler struct {
	mu sync.Mutex
	//p           Provider
	sm          *SessionManager
	readTimeout time.Duration
}

// NewHandler creates a new Handler given a SQLe engine.
func NewHandler(sm *SessionManager, rt time.Duration) *Handler {
	return &Handler{
		sm:          sm,
		readTimeout: rt,
	}
}

// NewConnection is called when a connection is created.
// It is not established yet. The handler can decide to
// set StatusFlags that will be returned by the handshake methods.
// In particular, ServerStatusAutocommit might be set.
func (h *Handler) NewConnection(c *mysql.Conn) {
	logrus.WithField(log.ConnectionIdLogField, c.ConnectionID).Infof("New connection")
}

// ConnectionReady is called after the connection handshake, but
// before we begin to process commands.
func (h *Handler) ConnectionReady(c *mysql.Conn) {
	//logrus.WithField(log.ConnectionIdLogField, c.ConnectionID).Infof("ConnectionReady")
}

// ConnectionClosed is called when a connection is closed.
func (h *Handler) ConnectionClosed(c *mysql.Conn) {
	ctx, _ := h.sm.NewContextWithQuery(c, "")
	h.sm.CloseConn(c)
	ctx.ClearWarnings()
	logrus.WithField(log.ConnectionIdLogField, c.ConnectionID).WithField(log.ConnectTimeLogKey, ctx.CreateTime().Format("2006-01-02 15:04:05.000000")).Infof("Connection closed")
}

// ComQuery is called when a connection receives a query.
// Note the contents of the query slice may change after
// the first call to callback. So the Handler should not
// hang on to the byte slice.
func (h *Handler) ComQuery(c *mysql.Conn, query string, callback func(*sqltypes.Result) error) error {
	return h.errorWrappedDoQuery(c, query, nil, callback)
}

// ComPrepare is called when a connection receives a prepared
// statement query.
func (h *Handler) ComPrepare(c *mysql.Conn, query string, bindVars map[string]*querypb.BindVariable) ([]*querypb.Field, error) {
	ctx, err := h.sm.NewContextWithQuery(c, query)
	if err != nil {
		return nil, err
	}
	logrus.Info(ctx)
	return nil, nil

}

// ComStmtExecute is called when a connection receives a statement
// execute query.
func (h *Handler) ComStmtExecute(c *mysql.Conn, prepare *mysql.PrepareData, callback func(*sqltypes.Result) error) error {
	return h.errorWrappedDoQuery(c, prepare.PrepareStmt, prepare.BindVars, callback)
}

// ComBinlogDumpGTID is called when a connection receives a ComBinlogDumpGTID request
func (h *Handler) ComBinlogDumpGTID(c *mysql.Conn, gtidSet mysql.GTIDSet) error {
	return nil
}

// WarningCount is called at the end of each query to obtain
// the value to be returned to the client in the EOF packet.
// Note that this will be called either in the context of the
// ComQuery callback if the result does not contain any fields,
// or after the last ComQuery call completes.
func (h *Handler) WarningCount(c *mysql.Conn) uint16 {
	if sess := h.sm.session(c); sess != nil {
		return sess.WarningCount()
	}

	return 0
}

func (h *Handler) ComResetConnection(c *mysql.Conn) {
	// TODO: handle reset logic
}

func setConnStatusFlags(ctx *engine.Context, c *mysql.Conn) error {
	//ok, err := isSessionAutocommit(ctx)
	//if err != nil {
	//	return err
	//}
	//if ok {
	//	c.StatusFlags |= uint16(mysql.ServerStatusAutocommit)
	//} else {
	//	c.StatusFlags &= ^uint16(mysql.ServerStatusAutocommit)
	//}

	return nil
}

// Call doQuery and cast known sys to SQLError
func (h *Handler) errorWrappedDoQuery(c *mysql.Conn, query string, bindVars map[string]*querypb.BindVariable, callback func(*sqltypes.Result) error) error {
	err := h.doQuery(c, query, bindVars, callback)
	err, ok := errs.CastSQLError(err)
	if ok {
		return nil
	} else {
		return err
	}
}

var queryLoggingRegex = regexp.MustCompile(`[\r\n\t ]+`)

func (h *Handler) doQuery(c *mysql.Conn, query string, bindVars map[string]*querypb.BindVariable, callback func(*sqltypes.Result) error) error {
	sys.QuestionIncrease()
	ctx, err := h.sm.NewContextWithQuery(c, query)
	if err != nil {
		return err
	}
	provider := ctx.GetCurrentDatabase()
	handled, err := h.handleKill(ctx, c, query)
	if err != nil {
		return err
	}

	if handled {
		return callback(&sqltypes.Result{})
	}

	ctx.SetLogger(ctx.GetLogger().WithField("query", string(queryLoggingRegex.ReplaceAll([]byte(query), []byte(" ")))))
	ctx.GetLogger().Debugf("Starting query")

	finish := observeQuery(ctx, query)
	defer finish(err)
	query = ctx.Sql()
	stmt, err := parser.Parse(query)
	if err != nil {
		return err
	}
	var r *sqltypes.Result
	switch s := stmt.(type) {
	case *parser.SelectStatement:
		if s.IsDual() {
			r, err := selectDual(ctx, s)
			if err != nil {
				return err
			}
			return callback(r)
		} else {
			return provider.Select(ctx, bindVars, callback, s)
		}
	case *parser.InsertStatement:
		r, err = provider.Insert(ctx, s)
		if err != nil {
			return err
		}
		callback(r)
	case *parser.DeleteStatement:
		r, err = provider.Delete(ctx, s)
		if err != nil {
			return err
		}
		callback(r)
	case *parser.UpdateStatement:
		r, err = provider.Update(ctx, s)
		if err != nil {
			return err
		}
		callback(r)
	case *parser.ShowStatement:
		switch s.Command {
		case "databases":
			r = showDatabases(s)
		case "tables":
			r, err = provider.ShowTables(ctx, s)
		case "status":
			r, err = showStatus(h.sm, s)
		case "charset":
			r, err = provider.ShowCharset(ctx, s)
		case "collation":
			r, err = provider.ShowCollation(ctx, s)
		case "variables":
			r, err = showVariables(ctx, s)
		default:
			r = &sqltypes.Result{}
		}
		if err != nil {
			return err
		}
		return callback(r)
	case *parser.SetStatement:
		r, err = setHandle(ctx, s)
		if err != nil {
			return err
		}
		return callback(r)
	case *parser.DescStatement:
		r, err := provider.DescTable(ctx, s)
		if err != nil {
			return err
		}
		return callback(r)
	case *parser.UseStatement:
		r, err = useHandle(h.sm, c, s)
		if err != nil {
			return err
		}
		callback(r)
	case *parser.AlterTableStatement:
		r, err = provider.AlterTable(ctx, s)
		if err != nil {
			return err
		}
		callback(r)
	case *parser.DropTableStatement:
		r, err = provider.DropTable(ctx, s)
		if err != nil {
			return err
		}
		callback(r)
	default:
		return fmt.Errorf("unsupported syntax: %#v", s)
	}
	return nil
}

func selectDual(ctx *engine.Context, stmt *parser.SelectStatement) (*sqltypes.Result, error) {
	r := &sqltypes.Result{
		Fields: []*querypb.Field{},
		Rows:   [][]sqltypes.Value{},
	}
	variables := ctx.GetAllEnvironmentVariables()
	row := []sqltypes.Value{}
	for _, field := range stmt.Fields {
		v, err := expr.Eval(field.Expr(), variables)
		if err != nil {
			msg := strings.ReplaceAll(err.Error(), parser.Sys_Variable_Prefix+".", "@@")
			return nil, fmt.Errorf(strings.ReplaceAll(msg, parser.User_Variable_Prefix+".", "@"))
		}
		var typ querypb.Type
		switch v.(type) {
		case int, int64:
			typ = sqltypes.Int64
		case string:
			typ = sqltypes.VarChar
		case int8:
			typ = sqltypes.Int8
		case int16:
			typ = sqltypes.Int16
		case int32:
			typ = sqltypes.Int32
		case float64:
			typ = sqltypes.Float64
		case float32:
			typ = sqltypes.Float32
		case decimal.Decimal:
			typ = sqltypes.Float64
		case bool:
			typ = sqltypes.Int8
			if v == true {
				v = 1
			} else {
				v = 0
			}
		}
		r.Fields = append(r.Fields, &querypb.Field{
			Name:    field.Name(),
			Type:    typ,
			Charset: collations.CollationUtf8ID,
		})
		row = append(row, sqltypes.MakeTrusted(typ, []byte(cast.ToString(v))))
	}
	r.RowsAffected = 1
	r.Rows = append(r.Rows, row)
	return r, nil
}

func showDatabases(stmt *parser.ShowStatement) *sqltypes.Result {
	schema := stmt.Node().Schema()
	r := &sqltypes.Result{
		Fields: schema.ToFields(),
	}
	for _, db := range conf.Databases {
		row := schema.CreateRow()
		typ := strings.ToUpper(db.Type)
		if typ == "" {
			typ = "DATABROKER"
		}
		row.Append(db.Name, typ, db.Comment)
		r.Rows = append(r.Rows, row.ToSQL())
	}
	r.RowsAffected = uint64(len(conf.Databases))
	return r
}

func showVariables(ctx *engine.Context, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	schema := stmt.Node().Schema()
	r := &sqltypes.Result{
		Fields: schema.ToFields(),
		Rows:   [][]sqltypes.Value{},
	}
	variables := ctx.GetAllUserVariables()
	reg := "^" + strings.ReplaceAll(stmt.Filter, "%", "\\w*") + "$"
	for k, v := range variables {
		matched, _ := regexp.MatchString(reg, k)
		if stmt.Filter == "" || matched {
			row := schema.CreateRow()
			row.Append(k, v)
			r.Rows = append(r.Rows, row.ToSQL())
		}
	}
	r.RowsAffected = uint64(len(r.Rows))
	return r, nil
}

func showStatus(sm *SessionManager, stmt *parser.ShowStatement) (*sqltypes.Result, error) {
	schema := stmt.Node().Schema()
	r := &sqltypes.Result{
		Fields: schema.ToFields(),
	}
	reg := "^" + strings.ReplaceAll(stmt.Filter, "%", "\\w*") + "$"
	matched, _ := regexp.MatchString(reg, "max_connections")
	if stmt.Filter == "" || matched {
		row := schema.CreateRow()
		row.Append("max_connections", cast.ToString(config.GetConfig().MaxConnections), "最大连接数")
		r.Rows = append(r.Rows, row.ToSQL())
	}

	matched, _ = regexp.MatchString(reg, "Threads_connected")
	if stmt.Filter == "" || matched {
		row := schema.CreateRow()
		row.Append("Threads_connected", cast.ToString(len(sm.sessions)), "打开的连接数")
		r.Rows = append(r.Rows, row.ToSQL())
	}

	matched, _ = regexp.MatchString(reg, "Threads_created")
	if stmt.Filter == "" || matched {
		row := schema.CreateRow()
		row.Append("Threads_created", cast.ToString(sys.ThreadsCreated), "创建过的连接数")
		r.Rows = append(r.Rows, row.ToSQL())
	}

	matched, _ = regexp.MatchString(reg, "Questions")
	if stmt.Filter == "" || matched {
		row := schema.CreateRow()
		row.Append("Questions", sys.Questions, "发往服务器的查询数量")
		r.Rows = append(r.Rows, row.ToSQL())
	}

	matched, _ = regexp.MatchString(reg, "Uptime")
	if stmt.Filter == "" || matched {
		row := schema.CreateRow()
		row.Append("Uptime", sys.Uptime(), "服务器工作时长")
		r.Rows = append(r.Rows, row.ToSQL())
	}

	matched, _ = regexp.MatchString(reg, "CPU_usage")
	if stmt.Filter == "" || matched {
		row := schema.CreateRow()
		row.Append("CPU_usage", sys.CPUUsage(), "CPU使用率")
		r.Rows = append(r.Rows, row.ToSQL())
	}

	matched, _ = regexp.MatchString(reg, "Memory_usage")
	if stmt.Filter == "" || matched {
		row := schema.CreateRow()
		row.Append("Memory_usage", sys.MemoryUsage(), "内存使用率")
		r.Rows = append(r.Rows, row.ToSQL())
	}
	r.RowsAffected = uint64(len(r.Rows))
	return r, nil
}

func setHandle(ctx *engine.Context, stmt *parser.SetStatement) (*sqltypes.Result, error) {
	name := stmt.Name
	if strings.HasPrefix(name, "@@") {
		//设置系统变量
	} else if strings.HasPrefix(name, "@") {
		//设置用户变量
		name = name[1:]
		ctx.SetUserVariable(name, stmt.Value)
	} else {
		//设置列值
	}
	return &sqltypes.Result{
		RowsAffected: 1,
	}, nil
}

func useHandle(sm *SessionManager, conn *mysql.Conn, stmt *parser.UseStatement) (*sqltypes.Result, error) {
	err := sm.SetDB(conn, stmt.Target)
	return &sqltypes.Result{}, err
}

func (h *Handler) pollForClosedConnection(ctx *engine.Context, c *mysql.Conn, errChan chan error, quit chan struct{}) {
	tcpConn, ok := maybeGetTCPConn(c.GetRawConn())
	if !ok {
		ctx.GetLogger().Trace("Connection checker exiting, connection isn't TCP")
		return
	}
	logrus.Info(tcpConn)
}

func maybeGetTCPConn(conn net.Conn) (*net.TCPConn, bool) {
	wrap, ok := conn.(netutil.ConnWithTimeouts)
	if ok {
		conn = wrap.Conn
	}

	tcp, ok := conn.(*net.TCPConn)
	if ok {
		return tcp, true
	}

	return nil, false
}

func (h *Handler) handleKill(ctx *engine.Context, conn *mysql.Conn, query string) (bool, error) {
	q := strings.ToLower(query)
	// TODO: move this to parser, normal execution path
	s := regKillCmd.FindStringSubmatch(q)
	if s == nil {
		return false, nil
	}
	ctx.GetLogger().Info("killing query")
	id, err := strconv.ParseUint(s[2], 10, 32)
	if err != nil {
		return false, err
	}
	connID := uint32(id)
	ctx.GetLogger().Infof("Connection id: %d", connID)
	if s[1] != "query" {
		ctx.GetLogger().Info("kill connection")
		h.sm.CloseConn(conn)
		conn.Close()
	}
	return true, nil
}

var (
	// QueryCounter describes a metric that accumulates number of queries monotonically.
	QueryCounter = discard.NewCounter()

	// QueryErrorCounter describes a metric that accumulates number of failed queries monotonically.
	QueryErrorCounter = discard.NewCounter()

	// QueryHistogram describes a queries latency.
	QueryHistogram = discard.NewHistogram()
)

func observeQuery(ctx *engine.Context, query string) func(err error) {
	span, _ := ctx.Span("query", opentracing.Tag{Key: "query", Value: query})

	t := time.Now()
	return func(err error) {
		if err != nil {
			QueryErrorCounter.With("query", query, "error", err.Error()).Add(1)
		} else {
			QueryCounter.With("query", query).Add(1)
			QueryHistogram.With("query", query, "duration", "seconds").Observe(time.Since(t).Seconds())
		}

		span.Finish()
	}
}
